package cn.aijson.demo


import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object WordCountYarn {
  def main(args: Array[String]): Unit = {
    //判断输入输出目录
    if (args.length<2) {
      println("需要制定input和output路径")
      System.exit(1) //非0 表示非正常退出
    }

    //1\准备spark context
    var conf:SparkConf=new SparkConf()
      .setAppName("workdcount")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //2\读文件,生成分布式数据集RDD
    val lines:RDD[String] = sc.textFile(args(0))

    //3\数据转换和操作
    //对行按照空格进行切割，生成单词列表
    val words = lines.flatMap(_.split(" "))
    //每个单词生成键值对
    val words_key:RDD[(String,Int)] = words.map((_,1))
    //    分组聚合
    val result = words_key.reduceByKey(_ + _)

    //输出
    //result.foreach(println)
    //设置操作用户
    System.setProperty("HADOOP_USER_NAME","root")
    result.repartition(1).saveAsTextFile(args(1))

    //让程序暂存运行,便于查看web ui信息
    Thread.sleep(60*1000)

    sc.stop()
  }
}
